package com.fintech.pangu.actuate.metrics.rocketmq;

import com.fintech.pangu.actuate.metrics.util.MetricsUtil;
import com.fintech.pangu.rocketmq.metrics.RocketMQConsumeMessageContext;
import com.fintech.pangu.rocketmq.metrics.RocketMQMetricConstants;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.DistributionSummary;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Tags;
import org.springframework.util.StringUtils;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Description: RocketMQ消费端注册
 * @Author xiexiaole
 * @Date 2019/12/17 9:33
 */
public class RocketMQConsumeMetrics {


    public void getConsumeCountCounter(RocketMQConsumeMessageContext context){
        Counter counter = MetricsUtil.recordCounter("rocketmq.client.consume.total",
                "rocketmq client consume success, total count",
                convertToTags(context));
        counter.increment(context.getConsumeCount());
    }
    public AtomicInteger getConsumeDelayGauge(RocketMQConsumeMessageContext context){
        AtomicInteger consumeDelayCount = MetricsUtil.recordGauge("rocketmq.client.consume.delay",
                "rocketmq not consume count",
                convertToTags(context));
        return consumeDelayCount;
    }
    public void getConsumeCostTime(RocketMQConsumeMessageContext context){
        DistributionSummary summary = MetricsUtil.recordDistributionSummary("rocketmq.client.consume.timer",
                "rocketmq client consume success, timer",
                convertToTags(context), 0.5, 0.9, 0.95, 0.99);
        summary.record(context.getDuration());
    }
    public void getConsumeKernelCounter(RocketMQConsumeMessageContext context){
        Counter counter = MetricsUtil.recordCounter("rocketmq.client.kernel.consume.total",
                "rocketmq client kernel consume success, total count",
                convertToTags(context));
        counter.increment(context.getConsumeCount());
    }

    /**
     * RocketMQConsumeMessageContext转换为 Micrometer Tags
     */
    private Tags convertToTags(RocketMQConsumeMessageContext context) {
        Tags tags = Tags.empty();

        if(context.isSuccess()){
            tags = tags.and(Tag.of(RocketMQMetricConstants.Tags.SUCCESS, "Y"));
        }
        else {
            tags = tags.and(Tag.of(RocketMQMetricConstants.Tags.SUCCESS, "N"));
        }

        if(context.isRetry()){
            tags = tags.and(Tag.of(RocketMQMetricConstants.Tags.RETRY, "Y"));
        }
        else {
            tags = tags.and(Tag.of(RocketMQMetricConstants.Tags.RETRY, "N"));
        }

        if(StringUtils.hasText(context.getTopic())){
            tags = tags.and(Tag.of(RocketMQMetricConstants.Tags.TOPIC, context.getTopic()));
        }

        if(StringUtils.hasText(context.getConsumerGroup())){
            tags = tags.and(Tag.of(RocketMQMetricConstants.Tags.CONSUMER_GROUP, context.getConsumerGroup()));
        }

//        if(StringUtils.hasText(context.getMessageModel())){
//            tags = tags.and(Tag.of(RocketMQMetricConstants.Tags.MESSAGE_MODEL, context.getMessageModel()));
//        }

//        if(StringUtils.hasText(context.getConsumeMode())){
//            tags = tags.and(Tag.of(RocketMQMetricConstants.Tags.CONSUME_MODE, context.getConsumeMode()));
//        }

//        if(StringUtils.hasText(context.getConsumeStatus())){
//            tags = tags.and(Tag.of(RocketMQMetricConstants.Tags.CONSUME_STATUS, context.getConsumeStatus()));
//        }

        return tags;
    }
}
